package com.settlement.system.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.settlement.system.common.constant.KafkaConstants;
import com.settlement.system.common.constant.SystemConstants;
import com.settlement.system.common.enums.PlatFormEnum;
import com.settlement.system.common.util.DateUtils;
import com.settlement.system.common.util.NumberUtils;
import com.settlement.system.common.util.StringUtils;
import com.settlement.system.handler.disruptor.MessageVo;
import com.settlement.system.mapper.DyBillMapper;
import com.settlement.system.model.entity.DyBill;
import com.settlement.system.model.entity.KafkaSendErrorRecord;
import com.settlement.system.model.entity.SysImportErrorDetail;
import com.settlement.system.model.entity.SysStore;
import com.settlement.system.model.form.DyBillForm;
import com.settlement.system.model.query.DyBillQuery;
import com.settlement.system.model.vo.DyBillImportVo;
import com.settlement.system.service.*;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 抖音结算单 服务实现类
 * </p>
 *
 * @author huangrensen
 * @since 2024-05-30
 */
@Service
@RequiredArgsConstructor
public class DyBillServiceImpl extends ServiceImpl<DyBillMapper, DyBill> implements DyBillService {

    private final Logger logger = LoggerFactory.getLogger(DyBillServiceImpl.class);

    private final RedisTemplate redisTemplate;

    private final SysStoreService sysStoreService;

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final SnowflakeIdWorker snowflakeIdWorker = SnowflakeIdWorker.getInstance();

    private final SysImportErrorDetailService importErrorDetailService;

    private final KafkaSendErrorRecordService kafkaSendErrorRecordService;

    @Override
    public Page<DyBill> getPage(DyBillQuery queryParams) {
        // 查询参数
        int pageNum = queryParams.getPageNum();
        int pageSize = queryParams.getPageSize();
        // 拼接月份列表
        List<String> monthList = new ArrayList<>();
        String endMonthStr = queryParams.getEndMonth().replaceAll("-", "");
        LocalDate endMonth = LocalDate.parse(endMonthStr + "01", DateTimeFormatter.ofPattern("yyyyMMdd"));
        LocalDate startTime = LocalDate.parse(queryParams.getStartMonth().replaceAll("-", "") + "01", DateTimeFormatter.ofPattern(
                "yyyyMMdd"));
        // 获取已经生成的表名
        Object object = redisTemplate.opsForValue().get(SystemConstants.DY_BILL);
        List<String> tableNameList;
        if (object != null) {
            tableNameList = (List<String>) object;
        } else {
            tableNameList = this.baseMapper.getAllTable();
            // 添加缓存
            redisTemplate.opsForValue().set(SystemConstants.DY_BILL, tableNameList, 1, TimeUnit.HOURS);
        }
        // 循环追加月份
        while (!startTime.isAfter(endMonth)) {
            monthList.add(startTime.format(DateTimeFormatter.ofPattern("yyyyMM")));
            startTime = startTime.plusMonths(1);
        }
        // 根据月份判断数据库表是否存在
        List<String> existTableList = new ArrayList<>();
        String tableName;
        for (String month : monthList) {
            tableName = "dy_bill_%s".formatted(month);
            if (tableNameList.contains(tableName)) {
                existTableList.add(tableName);
            }
        }
        queryParams.setTableNameList(existTableList);
        // 查询数据
        Page<DyBill> page = new Page<>(pageNum, pageSize);
        if (existTableList.isEmpty()) {
            return page;
        }
        Page<DyBill> mapperPage = this.baseMapper.getPage(page, queryParams);
        SysStore sysStore;
        for (DyBill record : mapperPage.getRecords()) {
            sysStore = sysStoreService.getById(record.getStoreId());
            if (sysStore != null) {
                record.setStoreName(sysStore.getCommercePlatformStoreName());
            }
        }
        return mapperPage;
    }

    @Override
    public boolean saveDyBill(DyBillForm dyBillForm) {
        DyBill dyBill = new DyBill();
        BeanUtils.copyProperties(dyBillForm, dyBill);
        dyBill.setCreateTime(null);
        return this.saveOrUpdate(dyBill);
    }

    @Override
    public DyBillForm getDyBillFormData(Long id) {
        DyBill dyBill = this.getById(id);
        DyBillForm dyBillForm = new DyBillForm();
        BeanUtils.copyProperties(dyBill, dyBillForm);
        return dyBillForm;
    }

    @Override
    public boolean deleteDyBill(String ids) {
        Assert.isTrue(StringUtils.hasText(ids), "删除的数据为空");
        List<Long> idArray = Arrays.stream(ids.split(",")).map(Long::parseLong).collect(Collectors.toList());
        return this.removeByIds(idArray);
    }

    @Override
    public void importData(List<DyBillImportVo> voList) {
        List<String> billMonthList = voList.stream().map(DyBillImportVo::getBillMonth).distinct().toList();
        for (String billMonth : billMonthList) {
            if (!StringUtils.hasText(this.baseMapper.existTable("dy_bill_%s".formatted(billMonth)))) {
                this.baseMapper.createTable("dy_bill_%s".formatted(billMonth));
            }
        }
        List<DyBill> dyBillList = new ArrayList<>();
        for (DyBillImportVo importVo : voList) {
            DyBill dyBill = new DyBill();
            BeanUtils.copyProperties(importVo, dyBill);
            // 商品数量
            dyBill.setQuantity(NumberUtils.convertInteger(importVo.getQuantity()));
            // 达人补贴
            dyBill.setDrSubsidy(NumberUtils.convertBigDecimal(importVo.getDrSubsidy()));
            // 结算时间
            dyBill.setSettlementTime(DateUtils.convertStringToLocalDateTime(importVo.getSettlementTime()));
            // 结算金额
            dyBill.setSettlementPrice(NumberUtils.convertBigDecimal(importVo.getSettlementPrice()));
            // 下单时间
            dyBill.setOrderTime(DateUtils.convertStringToLocalDateTime(importVo.getOrderTime()));
            // 订单总价
            dyBill.setOrderAmount(NumberUtils.convertBigDecimal(importVo.getOrderAmount()));
            // 商品总价
            dyBill.setProductAmount(NumberUtils.convertBigDecimal(importVo.getProductAmount()));
            // 运费
            dyBill.setTransportFee(NumberUtils.convertBigDecimal(importVo.getTransportFee()));
            // 店铺劵
            dyBill.setPreferential(NumberUtils.convertBigDecimal(importVo.getPreferential()));
            // 结算前退款金额
            dyBill.setSettlementRefundAmount(NumberUtils.convertBigDecimal(importVo.getSettlementRefundAmount()));
            // 平台补贴
            dyBill.setPreformSubsidy(NumberUtils.convertBigDecimal(importVo.getPreformSubsidy()));
            // 达人佣金
            dyBill.setCommission(NumberUtils.convertBigDecimal(importVo.getCommission()));
            // 抖音支付补贴
            dyBill.setDySubsidy(NumberUtils.convertBigDecimal(importVo.getDySubsidy()));
            // 抖音月付营销补贴
            dyBill.setDyMonthSubsidy(NumberUtils.convertBigDecimal(importVo.getDyMonthSubsidy()));
            // 银行补贴
            dyBill.setBankSubsidy(NumberUtils.convertBigDecimal(importVo.getBankSubsidy()));
            // 以旧换新抵扣
            dyBill.setDeduction(NumberUtils.convertBigDecimal(importVo.getDeduction()));
            // 平台补贴运费
            dyBill.setPlatformSubsidyFee(NumberUtils.convertBigDecimal(importVo.getPlatformSubsidyFee()));
            // 用户实付
            dyBill.setBuyAmount(NumberUtils.convertBigDecimal(importVo.getBuyAmount()));
            // 收入合计
            dyBill.setReiceveAmount(NumberUtils.convertBigDecimal(importVo.getReiceveAmount()));
            // 平台服务费
            dyBill.setServiceAmount(NumberUtils.convertBigDecimal(importVo.getServiceAmount()));
            // 达人佣金
            dyBill.setCommission(NumberUtils.convertBigDecimal(importVo.getCommission()));
            // 服务商佣金
            dyBill.setCarrierCommission(NumberUtils.convertBigDecimal(importVo.getCarrierCommission()));
            // 渠道分成
            dyBill.setChannelFee(NumberUtils.convertBigDecimal(importVo.getChannelFee()));
            // 招商服务费
            dyBill.setInvestmentServiceFee(NumberUtils.convertBigDecimal(importVo.getInvestmentServiceFee()));
            // 站外推广费
            dyBill.setPromotionFee(NumberUtils.convertBigDecimal(importVo.getPromotionFee()));
            // 其他分成
            dyBill.setOtherFee(NumberUtils.convertBigDecimal(importVo.getOtherFee()));
            // 支出合计
            dyBill.setPayAmount(NumberUtils.convertBigDecimal(importVo.getPayAmount()));
            // 免佣金额
            dyBill.setNoCommissionAmount(NumberUtils.convertBigDecimal(importVo.getNoCommissionAmount()));
            dyBill.setStoreId(importVo.getStoreId());
            dyBill.setTableName("dy_bill_%s".formatted(importVo.getBillMonth()));
            dyBill.setId(snowflakeIdWorker.nextId(importVo.getBillMonth()));
            dyBillList.add(dyBill);
        }
        if (dyBillList.size() > 0) {
            dyBillList.forEach(item -> this.baseMapper.insertData(item));
        }
        // 删除缓存
        redisTemplate.delete(SystemConstants.DY_BILL);
        // 发送队列消息通知
        MessageVo vo;
        for (DyBill dyBill : dyBillList) {
            vo = new MessageVo();
            vo.setDataId(dyBill.getId());
            vo.setStoreId(dyBill.getStoreId());
            vo.setBillMonth(dyBill.getBillMonth());
            // 根据字典中的备注查找对应的数据库表字段名
            vo.setTableNamePrefix("dy_bill_");
            // 从字段管理里找到对应的字段类型编码，获取对应的billType
            vo.setDictCode(PlatFormEnum.DY_CONFIG.getValue());
            // 推送到转换标准账单的消息队列
            vo.setUuid(snowflakeIdWorker.nextId());
            CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_BILL_QUEUE, JSONObject.toJSONString(vo));
            MessageVo finalVo = vo;
            completableFuture.whenComplete((result, ex) -> {
                if (Objects.nonNull(ex)) {
                    ex.printStackTrace();
                    logger.info("推送DyBill数据到消息队列失败，失败原因：{}", ex.getMessage());
                    KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                    errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                    errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_BILL_QUEUE);
                    errorRecord.setRemark(ex.getMessage());
                    errorRecord.setUuid(finalVo.getUuid());
                    kafkaSendErrorRecordService.checkAndSave(errorRecord);
                }
            });
        }
    }

    @KafkaListener(id = KafkaConstants.HWG_TOPIC_DY_BILL, topics = KafkaConstants.HWG_TOPIC_DY_BILL, groupId = "yzConsumerGroup")
    public void importData(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
        logger.info("【dy-bill-queue】当前消费者本次拉取的数据量为：{}", (Objects.isNull(records)? 0 : records.size()));
        if(Objects.isNull(records) || records.isEmpty()){
            logger.error("【dy-bill-queue】无消费数据...");
        }
        String orderNo = "";
        for (ConsumerRecord<String, String> record : records) {
            String importVoStr = record.value();
            try{
                DyBillImportVo importVo = JSONObject.parseObject(importVoStr, DyBillImportVo.class);
                orderNo = importVo.getOrderNo();
                this.importData(Collections.singletonList(importVo));
            }catch (Exception e){
                logger.info("消费失败：{}", e.getMessage());
                // 保存失败数据
                SysImportErrorDetail detail = new SysImportErrorDetail();
                detail.setPlatformAndType("抖音账单");
                detail.setDataText(importVoStr);
                detail.setOrderNo(orderNo);
                detail.setQueueName(KafkaConstants.HWG_TOPIC_DY_BILL);
                detail.setRemark(e.getMessage());
                importErrorDetailService.checkAndCreate(detail);
            }
        }
        // 手动提交
        acknowledgment.acknowledge();
    }

    @Override
    public List<DyBillImportVo> findByIds(String idsStr) {
        if (!StringUtils.hasText(idsStr)) {
            return new ArrayList<>();
        }
        List<Long> ids = Arrays.stream(idsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<DyBill> dyBillList = this.getBaseMapper().selectBatchIds(ids);
        List<DyBillImportVo> voList = new ArrayList<>();
        for (DyBill dyBill : dyBillList) {
            DyBillImportVo vo = new DyBillImportVo();
            BeanUtils.copyProperties(dyBill, vo);
            voList.add(vo);
        }
        return voList;
    }
}
